package com.cike.sparkstudy.sql.java;

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.sql.DataFrame;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SQLContext;

import java.util.List;

/**
 * 使用反射方式将RDD转换成DataFrame
 */
public class RDD2DataFrameReflection {

    public static void main(String[] args){

        //创建普通的RDD
        SparkConf conf = new SparkConf()
                .setMaster("local")
                .setAppName("RDD2DataFrameReflection");
        JavaSparkContext sc = new JavaSparkContext(conf);
        SQLContext sqlContext = new SQLContext(sc);

        JavaRDD<String> lines = sc.textFile("/developerCodes/test/students.txt");

        JavaRDD<Student> students = lines.map(new Function<String, Student>() {
            @Override
            public Student call(String line) throws Exception {
                String[] lineSplit = line.split(",");
                Student stu = new Student();
                stu.setId(Integer.valueOf(lineSplit[0].trim()));
                stu.setName(lineSplit[1].trim());
                stu.setAge(Integer.valueOf(lineSplit[2].trim()));

                return stu;
            }
        });

        /**
         * 使用反射方式把RDD转换成DataFrame
         *  1、把student.class传入进去，就是用反射的方式创建DataFrame,因为Student.class本身就是反射的应用
         *  2、底层把Student.class进行反射，获取Student的fileId
         *          注意：这里的Student必须实现序列化接口Serializable接口
         */

        DataFrame studentDF = sqlContext.createDataFrame(students, Student.class);

        //拿到dataframe后，可以注册为一个临时表，然后针对这个临时表数据执行SQL
        studentDF.registerTempTable("students");

        //针对这个临时表数据执行SQL，查询年龄小于18岁的学生，就是teenageer
        DataFrame teenageerDF = sqlContext.sql("select * from students where age <= 18");

        //将查询出来的DataFrame转换成RDD
        JavaRDD<Row> teenageerRDD = teenageerDF.toJavaRDD();

        //将RDD中的数据，进行映射，映射为Student
        JavaRDD<Student> teenageStudentRDD = teenageerRDD.map(new Function<Row, Student>() {
            @Override
            public Student call(Row row) throws Exception {
                Student stu = new Student();
                stu.setAge(row.getInt(0));
                stu.setId(row.getInt(1));
                stu.setName(row.getString(2));
                return stu;
            }
        });

        //把数据打印出来
        List<Student> studentList = teenageStudentRDD.collect();
        for (Student stu : studentList){
            System.out.println(stu);
        }

    }



}
